package com.liwenfeng;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.*;

public class MyStream<T> {
    private final Collection<T> collection;
    private boolean parallel = false;
    private boolean isModify = false;

    // 构造方法
    public MyStream(Collection<T> collection) {
        if (collection.isEmpty()) {
            throw new IllegalArgumentException();
        }
        this.collection = collection;
    }

    public MyStream(Collection<T> collection, boolean parallel) {
        if (collection.isEmpty()) {
            throw new IllegalArgumentException();
        }
        this.collection = collection;
        this.parallel = parallel;
    }

    // 静态方法
    public static <T> MyStream<T> of(Collection<T> collection) {
        return new MyStream<>(collection);
    }

    @SafeVarargs
    public static <T> MyStream<T> of(T... ts) {
        return new MyStream<>(List.of(ts));
    }

    // 过滤
    public MyStream<T> filter(Predicate<T> predicate) {
        isModify = isModify();
        Collection<T> collection1 = new ArrayList<>();
        for (T t : collection) {
            if (predicate.test(t)) {
                collection1.add(t);
            }
        }
        return new MyStream<>(collection1, parallel);
    }

    // 映射
    public <R> MyStream<R> map(Function<T, R> function) {
        isModify = isModify();
        Collection<R> collection1 = new ArrayList<>();
        ArrayList<CompletableFuture<R>> completableFutures = new ArrayList<>();
        for (T t : collection) {
            if (parallel) {
                completableFutures.add(CompletableFuture.supplyAsync(() -> function.apply(t)));
            } else {
                collection1.add(function.apply(t));
            }
        }
        if (parallel) {
            for (CompletableFuture<R> future : completableFutures) {
                try {
                    collection1.add(future.get());
                } catch (InterruptedException | ExecutionException e) {
                    throw new RuntimeException(e);
                }
            }
        }
        return new MyStream<>(collection1, parallel);
    }

    // 排序
    public MyStream<T> sorted(Comparator<T> comparator) {
        isModify = isModify();
        List<T> collection1 = new ArrayList<>(collection);
        collection1.sort(comparator);
        return new MyStream<>(collection1, parallel);
    }

    // 截断
    public MyStream<T> limit(Integer count) {
        isModify = isModify();
        if (count < 0) throw new IllegalArgumentException();
        Collection<T> collection1 = new ArrayList<>();
        int i = collection.size() > count ? count : collection.size();
        Iterator<T> iterator = collection.iterator();
        for (; i > 0; i--) {
            collection1.add(iterator.next());
        }
        return new MyStream<>(collection1, parallel);
    }

    // 跳过
    public MyStream<T> skip(Integer count) {
        isModify = isModify();
        if (count < 0) throw new IllegalArgumentException();
        Collection<T> collection1 = new ArrayList<>();
        int i = collection.size() > count ? count : collection.size();
        Iterator<T> iterator = collection.iterator();
        for (int j = i; j > 0; j--) {
            iterator.next();
        }
        for (int j = i; j < collection.size(); j++) {
            collection1.add(iterator.next());
        }
        return new MyStream<>(collection1, parallel);
    }

    // 去重
    public MyStream<T> distinct() {
        isModify = isModify();
        Collection<T> collection1 = new ArrayList<>();
        for (T t : collection) {
            if (!collection1.contains(t)) {
                collection1.add(t);
            }
        }
        return new MyStream<>(collection1, parallel);
    }

    // 并行
    public MyStream<T> parallel() {
        parallel = true;
        return this;
    }

    public MyStream<T> parallel(boolean isParallel) {
        parallel = isParallel;
        return this;
    }

    // 摊开
    public <R> MyStream<R> flatMap(Function<T, MyStream<R>> function) {
        isModify = isModify();
        Collection<R> collection1 = new ArrayList<>();
        for (T t : collection) {
            MyStream<R> rs = function.apply(t);
            rs.forEach(collection1::add);
        }
        return new MyStream<>(collection1, parallel);
    }

    // 遍历
    public void forEach(Consumer<T> consumer) {
        isModify = isModify();
        for (T t : collection) {
            if (parallel) {
                new Thread(() -> consumer.accept(t)).start();
            } else {
                consumer.accept(t);
            }
        }
    }

    // 聚合
    public T reduce(T t1, BinaryOperator<T> operator) {
        isModify = isModify();
        T p = t1;
        for (T t : collection) {
            p = operator.apply(t, p);
        }
        return p;
    }

    public T reduce(BinaryOperator<T> operator) {
        return reduce(collection.iterator().next(), operator);
    }

    // 收集
    public <R> Collection<R> collect(Supplier<Collection<R>> supplier, BiConsumer<Collection<R>, T> consumer) {
        isModify = isModify();
        Collection<R> collection1 = supplier.get();
        for (T t : collection) {
            consumer.accept(collection1, t);
        }
        return collection1;
    }

    // 判断是否修改过
    private boolean isModify() {
        if (isModify) {
            throw new IllegalStateException("myStream has already been operated upon or closed");
        }
        return true;
    }
}
